package com.spark.service;

import com.alibaba.fastjson.JSONObject;
import com.spark.dao.ISessionAggrStatDAO;
import com.spark.dao.TaskDao;
import com.spark.daoImpl.DaoFactory;
import com.spark.pojo.SessionAggrStat;
import com.spark.pojo.Task;
import com.spark.utils.*;
import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import scala.Tuple2;

import java.sql.SQLException;
import java.util.Date;
import java.util.Iterator;

/**
 * @author jianger
 * @ date  2018/3/30-11:03
 * @project com.spark.service
 */
public class UserVisitSessionAnalyzeSpark {

    public static void main(String[] args) throws SQLException {
        args = new String[]{"1"};
        // 构建sparkConf
        SparkConf sparkConf = new SparkConf().setAppName(ConfigurationManager.getProperty(Constants.SPARK_CONF_APPNAME)).setMaster("local");

        // 构建spark上下文
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        //构建sparksql
        SQLContext sqlContext = getSqlContext(sc.sc());
        //生成测试数据
        mockData(sc, sqlContext);
        // 生成所有dao组件
        TaskDao daoImpl = DaoFactory.getDaoImpl();

        // 首先查询指定的任务,并获取任务的行为参数
        long taskId = ParamUtils.getTaskIdFromArgs(args);
        Task task = daoImpl.findById(taskId);

        System.out.println("task为" + task.toString());

        JSONObject taskParam = JSONObject.parseObject(task.getTask_param());
        // 如果要进行session粒度的数据聚合
        // 首先要从user_visit_action表中，查询出来指定日期范围内的行为数据
        JavaRDD<Row> actionRdd = getActionRDDByDateRange(sqlContext, taskParam);
        JavaPairRDD<String, Row> sessionToaction = getSessionid2ActionRDD(actionRdd);

        // 首先，可以将行为数据，按照session_id进行groupByKey分组
        // 此时的数据的粒度就是session粒度了，然后呢，可以将session粒度的数据
        // 与用户信息数据，进行join
        // 然后就可以获取到session粒度的数据，同时呢，数据里面还包含了session对应的user的信息
        // 到这里为止，获取的数据是<sessionid,(sessionid,searchKeywords,clickCategoryIds,age,professional,city,sex)>
        JavaPairRDD<String, String> sessionid2AggrInfoRDD =
                aggregateBySession(sqlContext, actionRdd);

        long count = sessionid2AggrInfoRDD.count();
        System.out.println(count);

        // 接着，就要针对session粒度的聚合数据，按照使用者指定的筛选参数进行数据过滤
        // 相当于我们自己编写的算子，是要访问外面的任务参数对象的
        // 所以，大家记得我们之前说的，匿名内部类（算子函数），访问外部对象，是要给外部对象使用final修饰的

        // 重构，同时进行过滤和统计
/*

        Accumulator<String> sessionAggrStatAccumulator = sc.accumulator(
                "", new SessionAggrStatAccumulator());

        JavaPairRDD<String, String> filteredSessionid2AggrInfoRDD = filterSessionAndAggrStat(
                sessionid2AggrInfoRDD, taskParam, sessionAggrStatAccumulator);

        List<Tuple2<String, String>> take = filteredSessionid2AggrInfoRDD.take(10);
        for (Tuple2<String, String> stringStringTuple2 : take) {
            System.out.println("年龄过滤后"+stringStringTuple2._2.toString());
        }

*/


        // 计算出各个范围的session占比，并写入MySQL
//        calculateAndPersistAggrStat(sessionAggrStatAccumulator.value(),
//                task.getTask_id());

        //关闭spark上下文
        sc.close();

    }


    private static void calculateAndPersistAggrStat(String value, Long task_id) {
        /**
         * 计算各session范围占比，并写入MySQL
         * @param value
         */
            // 从Accumulator统计串中获取值
            long session_count = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.SESSION_COUNT));

            long visit_length_1s_3s = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.TIME_PERIOD_1s_3s));
            long visit_length_4s_6s = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.TIME_PERIOD_4s_6s));
            long visit_length_7s_9s = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.TIME_PERIOD_7s_9s));
            long visit_length_10s_30s = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.TIME_PERIOD_10s_30s));
            long visit_length_30s_60s = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.TIME_PERIOD_30s_60s));
            long visit_length_1m_3m = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.TIME_PERIOD_1m_3m));
            long visit_length_3m_10m = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.TIME_PERIOD_3m_10m));
            long visit_length_10m_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.TIME_PERIOD_10m_30m));
            long visit_length_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.TIME_PERIOD_30m));

            long step_length_1_3 = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.STEP_PERIOD_1_3));
            long step_length_4_6 = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.STEP_PERIOD_4_6));
            long step_length_7_9 = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.STEP_PERIOD_7_9));
            long step_length_10_30 = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.STEP_PERIOD_10_30));
            long step_length_30_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.STEP_PERIOD_30_60));
            long step_length_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
                    value, "\\|", Constants.STEP_PERIOD_60));

            // 计算各个访问时长和访问步长的范围
            double visit_length_1s_3s_ratio = NumberUtils.formatDouble(
                    (double)visit_length_1s_3s / (double)session_count, 2);
            double visit_length_4s_6s_ratio = NumberUtils.formatDouble(
                    (double)visit_length_4s_6s / (double)session_count, 2);
            double visit_length_7s_9s_ratio = NumberUtils.formatDouble(
                    (double)visit_length_7s_9s / (double)session_count, 2);
            double visit_length_10s_30s_ratio = NumberUtils.formatDouble(
                    (double)visit_length_10s_30s / (double)session_count, 2);
            double visit_length_30s_60s_ratio = NumberUtils.formatDouble(
                    (double)visit_length_30s_60s / (double)session_count, 2);
            double visit_length_1m_3m_ratio = NumberUtils.formatDouble(
                    (double)visit_length_1m_3m / (double)session_count, 2);
            double visit_length_3m_10m_ratio = NumberUtils.formatDouble(
                    (double)visit_length_3m_10m / (double)session_count, 2);
            double visit_length_10m_30m_ratio = NumberUtils.formatDouble(
                    (double)visit_length_10m_30m / (double)session_count, 2);
            double visit_length_30m_ratio = NumberUtils.formatDouble(
                    (double)visit_length_30m / (double)session_count, 2);

            double step_length_1_3_ratio = NumberUtils.formatDouble(
                    (double)step_length_1_3 / (double)session_count, 2);
            double step_length_4_6_ratio = NumberUtils.formatDouble(
                    (double)step_length_4_6 / (double)session_count, 2);
            double step_length_7_9_ratio = NumberUtils.formatDouble(
                    (double)step_length_7_9 / (double)session_count, 2);
            double step_length_10_30_ratio = NumberUtils.formatDouble(
                    (double)step_length_10_30 / (double)session_count, 2);
            double step_length_30_60_ratio = NumberUtils.formatDouble(
                    (double)step_length_30_60 / (double)session_count, 2);
            double step_length_60_ratio = NumberUtils.formatDouble(
                    (double)step_length_60 / (double)session_count, 2);

            // 将统计结果封装为Domain对象
            SessionAggrStat sessionAggrStat = new SessionAggrStat();
            sessionAggrStat.setTask_id(task_id);
            sessionAggrStat.setSession_count(session_count);
            sessionAggrStat.setS1_s3(visit_length_1s_3s_ratio);
            sessionAggrStat.setS4_s6(visit_length_4s_6s_ratio);
            sessionAggrStat.setS7_s9(visit_length_7s_9s_ratio);
            sessionAggrStat.setS10_s30(visit_length_10s_30s_ratio);
            sessionAggrStat.setS30_s60(visit_length_30s_60s_ratio);
            sessionAggrStat.setM1_m3(visit_length_1m_3m_ratio);
            sessionAggrStat.setM3_m10(visit_length_3m_10m_ratio);
            sessionAggrStat.setM10_m30(visit_length_10m_30m_ratio);
            sessionAggrStat.setM30(visit_length_30m_ratio);
            sessionAggrStat.setV1_3(step_length_1_3_ratio);
            sessionAggrStat.setV4_6(step_length_4_6_ratio);
            sessionAggrStat.setV7_9(step_length_7_9_ratio);
            sessionAggrStat.setV10_30(step_length_10_30_ratio);
            sessionAggrStat.setV30_60(step_length_30_60_ratio);
            sessionAggrStat.setV60(step_length_60_ratio);
            // 调用对应的DAO插入统计结果
            ISessionAggrStatDAO sessionAggrStatDAO = DaoFactory.getSessionAggrStatDAOImpl();
        try {
            sessionAggrStatDAO.save(sessionAggrStat);
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }


    /**
     * 根据用户数据进行过滤
     *
     * @param sessionid2AggrInfoRDD 聚合后的数据
     * @param taskParam             过滤参数
     * @param sessionAggrStatAccumulator
     * @return 过滤后的数据
     */
    private static JavaPairRDD<String, String>     filterSessionAndAggrStat(JavaPairRDD<String, String> sessionid2AggrInfoRDD, JSONObject taskParam, final Accumulator<String> sessionAggrStatAccumulator) {

        // 为了使用我们后面的ValieUtils，所以，首先将所有的筛选参数拼接成一个连接串
        // 此外，这里其实大家不要觉得是多此一举
        // 其实我们是给后面的性能优化埋下了一个伏笔
        String startAge = ParamUtils.getParam(taskParam, Constants.PARAM_START_AGE);
        String endAge = ParamUtils.getParam(taskParam, Constants.PARAM_END_AGE);
        String professionals = ParamUtils.getParam(taskParam, Constants.PARAM_PROFESSIONALS);
        String cities = ParamUtils.getParam(taskParam, Constants.PARAM_CITIES);
        String sex = ParamUtils.getParam(taskParam, Constants.PARAM_SEX);
        String keywords = ParamUtils.getParam(taskParam, Constants.PARAM_KEYWORDS);
        String categoryIds = ParamUtils.getParam(taskParam, Constants.PARAM_CATEGORY_IDS);
        String _parameter = (startAge != null ? Constants.PARAM_START_AGE + "=" + startAge + "|" : "")
                + (endAge != null ? Constants.PARAM_END_AGE + "=" + endAge + "|" : "")
                + (professionals != null ? Constants.PARAM_PROFESSIONALS + "=" + professionals + "|" : "")
                + (cities != null ? Constants.PARAM_CITIES + "=" + cities + "|" : "")
                + (sex != null ? Constants.PARAM_SEX + "=" + sex + "|" : "")
                + (keywords != null ? Constants.PARAM_KEYWORDS + "=" + keywords + "|" : "")
                + (categoryIds != null ? Constants.PARAM_CATEGORY_IDS + "=" + categoryIds : "");
        if (_parameter.endsWith("\\|")) {
            _parameter = _parameter.substring(0, _parameter.length() - 1);
        }
        final String parameter = _parameter;

        // 根据参数过滤数据
        JavaPairRDD<String, String> aggrInfofilter = sessionid2AggrInfoRDD.filter(new Function<Tuple2<String, String>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, String> tuple2) throws Exception {

                // 获取聚合数据
                String aggrInfo = tuple2._2;
                // 根据条件，组个判断

                // 接着，依次按照筛选条件进行过滤
                // 按照年龄范围进行过滤（startAge、endAge）
                if (!ValidUtils.between(aggrInfo, Constants.FIELD_AGE, parameter, Constants.PARAM_START_AGE, Constants.PARAM_END_AGE)) {
                    return false;
                }
                // 按照职业范围进行过滤（professionals）
                // 互联网,IT,软件
                // 互联网
                if (!ValidUtils.in(aggrInfo, Constants.FIELD_PROFESSIONAL,
                        parameter, Constants.PARAM_PROFESSIONALS)) {
                    return false;
                }

                // 按照城市范围进行过滤（cities）
                // 北京,上海,广州,深圳
                // 成都
                if (!ValidUtils.in(aggrInfo, Constants.FIELD_CITY,
                        parameter, Constants.PARAM_CITIES)) {
                    return false;
                }

                // 按照性别进行过滤
                // 男/女
                // 男，女
                if (!ValidUtils.equal(aggrInfo, Constants.FIELD_SEX,
                        parameter, Constants.PARAM_SEX)) {
                    return false;
                }

                // 按照搜索词进行过滤
                // 我们的session可能搜索了 火锅,蛋糕,烧烤
                // 我们的筛选条件可能是 火锅,串串香,iphone手机
                // 那么，in这个校验方法，主要判定session搜索的词中，有任何一个，与筛选条件中
                // 任何一个搜索词相当，即通过
                if (!ValidUtils.in(aggrInfo, Constants.FIELD_SEARCH_KEYWORDS,
                        parameter, Constants.PARAM_KEYWORDS)) {
                    return false;
                }
                // 按照点击品类id进行过滤
                if (!ValidUtils.in(aggrInfo, Constants.FIELD_CLICK_CATEGORY_IDS,
                        parameter, Constants.PARAM_CATEGORY_IDS)) {
                    return false;
                }
                // 如果经过了之前的多个过滤条件之后，程序能够走到这里
                // 那么就说明，该session是通过了用户指定的筛选条件的，也就是需要保留的session
                // 那么就要对session的访问时长和访问步长，进行统计，根据session对应的范围
                // 进行相应的累加计数

                // 主要走到这一步，那么就是需要计数的session
                sessionAggrStatAccumulator.add(Constants.SESSION_COUNT);

                // 计算出session的访问时长和访问步长的范围，并进行相应的累加
                long visitLength = Long.valueOf(StringUtils.getFieldFromConcatString(
                        aggrInfo, "\\|", Constants.FIELD_VISIT_LENGTH));
                long stepLength = Long.valueOf(StringUtils.getFieldFromConcatString(
                        aggrInfo, "\\|", Constants.FIELD_STEP_LENGTH));
                calculateVisitLength(visitLength);
                calculateStepLength(stepLength);
                return true;
            }

            /**
             * 计算访问时长范围
             * @param visitLength
             */
            private void calculateVisitLength(long visitLength) {
                if(visitLength >=1 && visitLength <= 3) {
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1s_3s);
                } else if(visitLength >=4 && visitLength <= 6) {
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_4s_6s);
                } else if(visitLength >=7 && visitLength <= 9) {
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_7s_9s);
                } else if(visitLength >=10 && visitLength <= 30) {
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10s_30s);
                } else if(visitLength > 30 && visitLength <= 60) {
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30s_60s);
                } else if(visitLength > 60 && visitLength <= 180) {
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_1m_3m);
                } else if(visitLength > 180 && visitLength <= 600) {
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_3m_10m);
                } else if(visitLength > 600 && visitLength <= 1800) {
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_10m_30m);
                } else if(visitLength > 1800) {
                    sessionAggrStatAccumulator.add(Constants.TIME_PERIOD_30m);
                }
            }

            /**
             * 计算访问步长范围
             * @param stepLength
             */
            private void calculateStepLength(long stepLength) {
                if(stepLength >= 1 && stepLength <= 3) {
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_1_3);
                } else if(stepLength >= 4 && stepLength <= 6) {
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_4_6);
                } else if(stepLength >= 7 && stepLength <= 9) {
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_7_9);
                } else if(stepLength >= 10 && stepLength <= 30) {
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_10_30);
                } else if(stepLength > 30 && stepLength <= 60) {
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_30_60);
                } else if(stepLength > 60) {
                    sessionAggrStatAccumulator.add(Constants.STEP_PERIOD_60);
                }
            }

        });
        return aggrInfofilter;
    }

    /**
     * 构建一个获取sparksql的方法
     * 如果是本地模式就sqlcontext对象
     * 如果是在生产环境，就使用hivecontext对象
     */

    private static SQLContext getSqlContext(SparkContext sc) {

        boolean local = ConfigurationManager.getBoolean(Constants.LOCAL_FLAGE);
        if (local) {

            return new SQLContext(sc);

        } else {
            return new HiveContext(sc);
        }
    }

    /**
     * 生成模拟数据（只有本地模式，才会去生成模拟数据）
     *
     * @param sc
     * @param sqlContext
     */
    private static void mockData(JavaSparkContext sc, SQLContext sqlContext) {
        boolean local = ConfigurationManager.getBoolean(Constants.LOCAL_FLAGE);
        if (local) {
            MockData.mock(sc, sqlContext);
        }
    }


    /**
     * 获取指定日期范围内的用户访问行为数据
     *
     * @param sqlContext SQLContext
     * @param taskParam  任务参数
     * @return 行为数据RDD
     */
    private static JavaRDD<Row> getActionRDDByDateRange(
            SQLContext sqlContext, JSONObject taskParam) {
        String startDate = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE);
        String endDate = ParamUtils.getParam(taskParam, Constants.PARAM_END_DATE);

        String sql =
                "select * "
                        + "from user_visit_action "
                        + "where date>='" + startDate + "' "
                        + "and date<='" + endDate + "'";
        DataFrame actionDF = sqlContext.sql(sql);

        return actionDF.javaRDD();
    }


    /**
     * 获取sessionid2到访问行为数据的映射的RDD
     *
     * @param actionRDD
     * @return
     */
    private static JavaPairRDD<String, Row> getSessionid2ActionRDD(JavaRDD<Row> actionRDD) {
        // 进行seeionid的映射
        // 第一个参数是输入参数，是行为数据
        //第二，三个参数是输出参数，分别是sessionid和行为数据
        return actionRDD.mapToPair(new PairFunction<Row, String, Row>() {
            @Override
            public Tuple2<String, Row> call(Row row) throws Exception {
                return new Tuple2<String, Row>(row.getString(2), row);
            }
        });
    }


    /**
     * 完成按session粒度到行为数据的聚合
     * 1. 完成用户行为数据的（userid,partInfo）的聚合
     * 2. 完成用户信息（userId，userInfo）的聚合
     * 3. 完成join的所有数据的聚合
     *
     * @param sqlContext sqlcontext
     * @param actionRDD  行为数据
     * @return
     */
    private static JavaPairRDD<String, String> aggregateBySession(SQLContext sqlContext, JavaRDD actionRDD) {

        // 现在actionRDD中的元素是Row，一个Row就是一行用户访问行为记录，比如一次点击或者搜索
        // 我们现在需要将这个Row映射成<sessionid,Row>的格式
        // 调用上面的映射函数
        JavaPairRDD sessionid2ActionRDD = getSessionid2ActionRDD(actionRDD);

        // 对这个数据按照sessioId进行分组

        JavaPairRDD<String, Iterable<Row>> javaPairRDD = sessionid2ActionRDD.groupByKey();
        // 对每一个session分组进行聚合，将session中所有的搜索词和点击品类都聚合起来
        // 到此为止，获取的数据格式，如下：<userid,partAggrInfo(sessionid,searchKeywords,clickCategoryIds)>
        // 1. 完成行为数据的聚合
        JavaPairRDD<Long, String> userid2PartAggrInfoRDD = javaPairRDD.mapToPair(new PairFunction<Tuple2<String, Iterable<Row>>, Long, String>() {
            @Override
            public Tuple2<Long, String> call(Tuple2<String, Iterable<Row>> tuple2) throws Exception {
                //1.获得userId
                Long userId = null;
                String sessionId = tuple2._1;
                Iterator<Row> iterabler = tuple2._2.iterator();
                StringBuffer searchKeywordsBuffer = new StringBuffer("");
                StringBuffer clickCategoryIdsBuffer = new StringBuffer("");


                Long userid = null;

                // session的起始和结束时间
                Date startTime = null;
                Date endTime = null;
                // session的访问步长
                int stepLength = 0;

                // 遍历row，获取user_id
                while (iterabler.hasNext()) {
                    Row row = iterabler.next();
                    if (userId == null) {
                        userId = row.getLong(1);
                    }

                    // 获取search_keyword
                    // 获取click_category_id
                    String search_keyword = row.getString(5);
                    Long click_category_id = row.getLong(6);
                    // 实际上这里要对数据说明一下
                    // 并不是每一行访问行为都有searchKeyword何clickCategoryId两个字段的
                    // 其实，只有搜索行为，是有searchKeyword字段的
                    // 只有点击品类的行为，是有clickCategoryId字段的
                    // 所以，任何一行行为数据，都不可能两个字段都有，所以数据是可能出现null值的

                    // 我们决定是否将搜索词或点击品类id拼接到字符串中去
                    // 首先要满足：不能是null值
                    // 其次，之前的字符串中还没有搜索词或者点击品类id

                    if (StringUtils.isNotEmpty(search_keyword)) {
                        if (!searchKeywordsBuffer.toString().contains(search_keyword)) {
                            searchKeywordsBuffer.append(search_keyword + ",");
                        }
                    }
                    if (click_category_id != null) {
                        if (!clickCategoryIdsBuffer.toString().contains(
                                String.valueOf(click_category_id))) {
                            clickCategoryIdsBuffer.append(click_category_id + ",");
                        }
                    }
                    // 计算session开始和结束时间
                    Date actionTime = DateUtils.parseTime(row.getString(4));

                    if (startTime == null) {
                        startTime = actionTime;
                    }
                    if (endTime == null) {
                        endTime = actionTime;
                    }

                    if (actionTime.before(startTime)) {
                        startTime = actionTime;
                    }
                    if (actionTime.after(endTime)) {
                        endTime = actionTime;
                    }

                    // 计算session访问步长
                    stepLength++;

                }


                // 计算session访问时长（秒）
                long visitLength = (endTime.getTime() - startTime.getTime()) / 1000;


                // 对搜索词和分类id进行去除空格
                String searchKeywords = StringUtils.trimComma(searchKeywordsBuffer.toString());
                String clickCategoryIds = StringUtils.trimComma(clickCategoryIdsBuffer.toString());

                // 拼接 partAggrInfo session的字符串

                String partAggrInfo = Constants.FIELD_SESSION_ID + "=" + sessionId + "|"
                        + (StringUtils.isNotEmpty(searchKeywords) ? Constants.FIELD_SEARCH_KEYWORDS + "=" + searchKeywords + "|" : "")
                        + (StringUtils.isNotEmpty(clickCategoryIds) ? Constants.FIELD_CLICK_CATEGORY_IDS + "=" + clickCategoryIds + "|" : "")
                        + Constants.FIELD_VISIT_LENGTH + "=" + visitLength + "|"
                        + Constants.FIELD_STEP_LENGTH + "=" + stepLength + "|"
                        + Constants.FIELD_START_TIME + "=" + DateUtils.formatTime(startTime);


                return new Tuple2<Long, String>(userId, partAggrInfo);
            }
        });


        //2. 完成用户数据的聚合

        // 查找用户数据

        String sql = "select * from user_info";

        DataFrame userData = sqlContext.sql(sql);
        JavaRDD<Row> userInfoRdd = userData.toJavaRDD();
        JavaPairRDD<Long, Row> userPair = userInfoRdd.mapToPair(new PairFunction<Row, Long, Row>() {
            @Override
            public Tuple2<Long, Row> call(Row row) throws Exception {
                return new Tuple2<Long, Row>(row.getLong(0), row);
            }
        });

        //3.对两个info进行聚合
        JavaPairRDD<Long, Tuple2<String, Row>> userFullInfoRdd = userid2PartAggrInfoRDD.join(userPair);
        // 对join起来的数据进行拼接，并且返回<sessionid,fullAggrInfo>格式的数据
        JavaPairRDD<String, String> sessionid2FullAggrInfoRDD = userFullInfoRdd.mapToPair(new PairFunction<Tuple2<Long, Tuple2<String, Row>>, String, String>() {
            @Override
            public Tuple2<String, String> call(Tuple2<Long, Tuple2<String, Row>> tupler) throws Exception {
                // 获取用户行为字符串
                String partInfo = tupler._2._1;
                //获取用户的信息row
                Row userRow = tupler._2._2;

                // 获取sessionId
                String sessionid = StringUtils.getFieldFromConcatString(
                        partInfo, "\\|", Constants.FIELD_SESSION_ID);

                int age = userRow.getInt(3);
                String professional = userRow.getString(4);
                String city = userRow.getString(5);
                String sex = userRow.getString(6);

                String fullAggrInfo = partInfo + "|"
                        + Constants.FIELD_AGE + "=" + age + "|"
                        + Constants.FIELD_PROFESSIONAL + "=" + professional + "|"
                        + Constants.FIELD_CITY + "=" + city + "|"
                        + Constants.FIELD_SEX + "=" + sex;

                return new Tuple2<String, String>(sessionid, fullAggrInfo);
            }
        });

        return sessionid2FullAggrInfoRDD;
    }


}
